package com.yanggu.flink.cep_api

import com.yanggu.flink.cep_api.pojo.LoginEvent
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._

import java.util

object LoginFailDetect {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    val keyByUserLoginDataStream = environment
      .fromElements(
        LoginEvent("user_1", "192.168.0.1", "fail", 2000L),
        LoginEvent("user_1", "192.168.0.2", "fail", 3000L),
        LoginEvent("user_2", "192.168.1.29", "fail", 4000L),
        LoginEvent("user_1", "171.56.23.10", "fail", 5000L),
        LoginEvent("user_2", "192.168.1.29", "success", 6000L),
        LoginEvent("user_2", "192.168.1.29", "fail", 7000L),
        LoginEvent("user_2", "192.168.1.29", "fail", 8000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[LoginEvent] {
          override def extractTimestamp(element: LoginEvent, recordTimestamp: Long) = element.timestamp
        }))
      .keyBy(_.userId)

    val pattern = Pattern
      .begin[LoginEvent]("first")
      .where(tmp => tmp.eventType.equals("fail"))
      .next("second")
      .where(tmp => tmp.eventType.equals("fail"))
      .next("third")
      .where(tmp => tmp.eventType.equals("fail"))

    val patternStream = CEP.pattern(keyByUserLoginDataStream, pattern)

    patternStream
      .select(new PatternSelectFunction[LoginEvent, String] {
        override def select(pattern: util.Map[String, util.List[LoginEvent]]) = {
          val first = pattern.get("first").get(0)
          val second = pattern.get("second").get(0)
          val third = pattern.get("third").get(0)
          s"user_id: ${first.userId}, 连续三次登录失败!登录时间: ${first.timestamp}, ${second.timestamp}, ${third.timestamp}"
        }
      })
      .print("warn")

    environment.execute()

  }

}
